package com.kili.recommend.data.config;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.kili.recommend.data.entity.Events;
import com.kili.recommend.data.entity.User;
import com.kili.recommend.data.service.EventsService;
import com.kili.recommend.data.service.UserService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.cache.Cache;
import org.springframework.cache.CacheManager;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.Map;
import java.util.Set;

/**
 * 接收kafka消息
 *
 * @author fenny
 * @since 2019/08/28
 */
@Slf4j
@Component
public class KafkaListenner {

    @Resource
    private CustomConfig customConfig;

    @Resource
    private UserService userService;

    @Resource
    private EventsService eventsService;

    @Resource
    private CacheManager cacheManager;

    @KafkaListener(topics = {"event_topic"})
    public void listen(String message) {
        try {
            JSONObject data = JSON.parseObject(message);
            // 对用户的标识，对未登录用户，可以填充设备标识、CookieID 等，对于登录用户，则应该填充注册账号；这里的例子，我们假设是一个未注册用户，所以填充的是一个设备编号；
            String distinctId = data.getString("distinct_id");
            if (distinctId.matches(IConstant.REG_NUMBER)) {
                Long userId = Long.parseLong(distinctId);
                if (userId > 0) {
                    String project = data.getString("project");
                    ProjectEnum projectEnum = ProjectEnum.restore(project);
                    if (projectEnum != ProjectEnum.other) {
                        //判断是否在采集范围
                        Map<String, Set<String>> manyCountryCollectEvent = customConfig.getGatherCountryAndEvent();
                        Set<String> countryEvents = manyCountryCollectEvent.get(projectEnum.toString());
                        String event = data.getString("event");
                        if (countryEvents != null && countryEvents.contains(event)) {
                            if (log.isDebugEnabled()) {
                                log.debug("msg={}", message);
                            }
                            //处理消息幂等
                            Long time = data.getLong("time");
                            String key = event + "_" + time + "_" + distinctId;
                            Cache cache = cacheManager.getCache(IConstant.CACHE_TRACK_ID);
                            if (cache != null && cache.putIfAbsent(key, System.currentTimeMillis()) == null) {
                                JSONObject properties = data.getJSONObject("properties");
                                String commodityId = properties.getString("commodityID");
                                if (commodityId != null && commodityId.matches(IConstant.REG_NUMBER)) {
                                    Long skuId = Long.parseLong(commodityId);
                                    User user = new User();
                                    user.setUserId(userId);
                                    userService.put(projectEnum, user);
                                    Events events = new Events();
                                    events.setEvent(event);
                                    events.setUserId(userId);
                                    events.setSkuId(skuId);
                                    events.setNum(1);
                                    eventsService.put(projectEnum, events);
                                }
                            }
                        }
                    }
                }
            }
        } catch (Exception e) {
            log.error(message, e);
        }
    }
}
